from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection


class CustomOperator(BaseOperator):
    def execute(self, context):
        gc = Connection.get("athena_default")
        print("The conn is", gc)
        # print("Conn uri is", gc.get_uri())


with DAG("example_get_connection_from_defn", schedule=None, catchup=False) as dag:
    CustomOperator(task_id="print_conn")

[2025-06-16, 17:55:11] INFO - 2025-06-16 12:25:11 [debug    ] DAG file parsed                [task] file=example-metadatadb-backend.py: chan="stdout": source="task"
[2025-06-16, 17:55:11] INFO - 2025-06-16 12:25:11 [debug    ] Connection not found in any of the configured Secrets Backends. Trying to retrieve from API server [task] conn_id=athena_default: chan="stdout": source="task"
[2025-06-16, 17:55:[11](http://localhost:28080/dags/example_get_connection_from_defn/runs/manual__2025-06-16T12:25:06.782393+00:00/tasks/print_conn?try_number=1#11)] INFO - The conn is Connection(conn_id='athena_default', conn_type='athena', description=None, host=None, schema=None, login=None, password=None, port=None, extra=None): chan="stdout": source="task"

DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-16, 18:06:[3](http://localhost:28080/dags/example_get_connection_from_defn/runs/manual__2025-06-16T12:36:33.447212+00:00/tasks/print_conn?try_number=1#3)3] INFO - Filling up the DagBag from /files/dags/example-metadatadb-backend.py: source="airflow.models.dagbag.DagBag"
[2025-06-16, 18:06:33] INFO - The conn is Connection(conn_id='athena_default', conn_type='athena', description=None, host=None, schema=None, login=None, password=None, port=None, extra=None): chan="stdout": source="task"